package stock

import (
	"bytes"
	"encoding/json"
	"errors"
	"io/ioutil"
	"learning/gooop/saga/mqs/logger"
	"learning/gooop/saga/mqs/models"
	"learning/gooop/saga/order"
	"net/http"
	"sync"
	"time"
)

type tStockService struct {
	rwmutex *sync.RWMutex
	stock map[string]int
	bMQReady bool
	publishQueue chan *models.TxMsg
}

func newStockService() IStockService {
	it := new(tStockService)
	it.init()
	return it
}


func (me *tStockService) init() {
	me.rwmutex = new(sync.RWMutex)
	me.stock = make(map[string]int)
	me.bMQReady = false
	me.publishQueue = make(chan *models.TxMsg, gMQMaxQueuedMsg)

	go func() {
		time.Sleep(100*time.Millisecond)
		go me.beginSubscribeMQ()
		go me.beginPublishMQ()
	}()
}

func (me *tStockService) GetStock(prodId string) int {
	me.rwmutex.RLock()
	defer me.rwmutex.RUnlock()

	it,ok := me.stock[prodId]
	if ok {
		return it
	} else {
		return 0
	}
}

func (me *tStockService) AddStock(prodId string, delta int) error {
	me.rwmutex.RLock()
	defer me.rwmutex.RUnlock()

	it,ok := me.stock[prodId]
	if ok {
		n := it + delta
		if n < 0 {
			logger.Logf("tStockService.AddStock, failed, prodId=%s, stock=%d, delta=%d", prodId, it, delta)
			return gInsufficientStockError
		} else {
			logger.Logf("tStockService.AddStock, done, prodId=%s, stock=%d, delta=%d, after=%d", prodId, it, delta, n)
			me.stock[prodId] = n
		}
	} else {
		if delta < 0 {
			logger.Logf("tStockService.AddStock, failed, prodId=%s, stock=0, delta=%d", prodId, delta)
			return gInsufficientStockError
		} else {
			logger.Logf("tStockService.AddStock, done, prodId=%s, stock=0, delta=%d, after=%d", prodId, it, delta)
			me.stock[prodId] = delta
		}
	}

	return nil
}


func (me *tStockService) beginSubscribeMQ() {
	expireDuration := int64(1 * time.Hour)
	subscribeDuration := 20 * time.Minute
	pauseDuration := 3*time.Second
	lastSubscribeTime := int64(0)

	for {
		now := time.Now().UnixNano()
		if now - lastSubscribeTime >= int64(subscribeDuration) {
			expireTime := now + expireDuration
			err := fnSubscribeMQ(expireTime)

			if err != nil {
				me.bMQReady = false
				logger.Logf("tStockService.beginSubscribeMQ, failed, err=%v", err)

			} else {
				lastSubscribeTime = now
				me.bMQReady = true
				logger.Logf("tStockService.beginSubscribeMQ, done")
			}
		}
		time.Sleep(pauseDuration)
	}
}

func fnSubscribeMQ(expireTime int64) error {
	msg := &models.SubscribeMsg{
		ClientID: gMQClientID,
		Topic: gMQSubscribeTopic,
		NotifyUrl: gMQServerURL + PathOfNotifySaleOrderCreated,
		ExpireTime: expireTime,
	}
	url := gMQServerURL + "/subscribe"
	return fnPost(msg, url)
}


func fnPost(msg interface{}, url string) error {
	body,_ := json.Marshal(msg)
	rsp, err := http.Post(url, "application/json;charset=utf-8", bytes.NewReader(body))
	if err != nil {
		return err
	}

	defer rsp.Body.Close()
	j, err := ioutil.ReadAll(rsp.Body)
	if err != nil {
		return err
	}
	ok := &models.OkMsg{}
	err = json.Unmarshal(j, ok)
	if err != nil {
		return err
	}

	if !ok.OK {
		return gMQReplyFalse
	}

	return nil
}



func (me *tStockService) beginPublishMQ() {
	for {
		select {
		case msg := <- me.publishQueue :
			me.publishMQ(msg)
			break
		}
	}
}

func (me *tStockService) publishMQ(msg *models.TxMsg) {
	url := gMQServerURL + "/publish"
	for i := 0;i < gMQMaxPublishRetry;i++ {
		err := fnPost(msg, url)
		if err != nil {
			logger.Logf("tStockService.publishMQ, failed, err=%v, msg=%v", err, msg)
			time.Sleep(gMQPublishInterval)

		} else {
			logger.Logf("tStockService.publishMQ, done, msg=%v", msg)
			return
		}
	}

	// publish failed
	logger.Logf("tStockService.publishMQ, failed max retries, msg=%v", msg)
}


func (me *tStockService) HandleSaleOrderCreated(it *order.SaleOrder) error {
	msg := &models.TxMsg{}
	msg.GlobalID = it.OrderID
	msg.SubID = it.OrderID + ".outbound"
	msg.SenderID = gMQClientID
	msg.Topic = gMQPublishTopic

	err := me.AddStock(it.ProductID, -it.Quantity)
	msg.CreateTime = time.Now().UnixNano()

	if err != nil {
		logger.Logf("tStockService.HandleSaleOrderCreated, err=%s, order=%v", err.Error(), it)
		msg.Content = "0"
	} else {
		msg.Content = "1"
	}

	if len(me.publishQueue) >= gMQMaxQueuedMsg {
		logger.Logf("tStockService.HandleSaleOrderCreated, err=%s, order=%v", gMQBlocked.Error(), it)
		return gMQBlocked

	} else {
		me.publishQueue <- msg
		return err
	}
}


var gInsufficientStockError = errors.New("insufficient stock")

var gMQReplyFalse = errors.New("mq reply false")
var gMQBlocked = errors.New("mq blocked")
var gMQMaxPublishRetry = 10
var gMQPublishInterval = 1*time.Second
var gMQSubscribeTopic = "sale-order.created"
var gMQPublishTopic = "sale-order.stock.outbound"
var gMQClientID = "stock-service"
var gMQServerURL = "http://localhost:3333"
var gMQMaxQueuedMsg = 1024

var MockStockService = newStockService()